package com.liuqi.knet.remoting.exchange.codec;

import com.liuqi.knet.remoting.exchange.Event;
import com.liuqi.knet.remoting.exchange.Request;
import com.liuqi.knet.remoting.exchange.Response;
import com.liuqi.knet.remoting.transport.AbstractCodec;
import com.liuqi.knet.serialize.Serialization;
import com.liuqi.knet.util.Bytes;
import com.liuqi.knet.util.CommonUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * 远程调用, 数据传输编解码
 *
 * header长度: 16位, header
 * [0 ~ 1]      MAGIC 固定标注位
 * [2]          FLAG  内容标记位: 1: 响应, 2: 请求, 3: 双向请求, 需要响应, 4: 事件
 * [3]          STATE
 *              <p>
 *                  {FLAG == 1} 响应, 为状态码
 *                  {FLAG == 2 || FLAG == 3}请求, 赋值为 0
 *                  {FLAG == 4} 事件, 为事件类型{@link Event.Type}
 *              </p>
 * [4 ~ 11]     消息ID
 * [12 ~ 15]    报文长度
 *
 * @author liuqi
 * @date 2025/3/6 13:20
 **/
public class ExchangeCodec extends AbstractCodec {

    /**
     * 请求头, 前两位标识头
     */
    private static final byte[] MAGIC = new byte[] {(byte) 0x10, (byte) 0x36};

    // ==========   内容标记位

    /**
     * FLAG标记位: 响应
     */
    private static final byte FLAG_RESPONSE = 0x01;

    /**
     * FLAG标记位: 请求
     */
    private static final byte FLAG_REQUEST = 0x02;

    /**
     * FLAG标记位: 请求, 需要服务端的响应
     */
    private static final byte FLAG_REQUEST_TWO_WAY = 0x03;

    /**
     * FLAG标记位: 事件
     */
    private static final byte FLAG_EVENT = 0x04;

    /**
     * 请求头长度
     */
    private static final int HEADER_LENGTH = 16;

    public ExchangeCodec(Serialization serialization) {
        super(serialization);
    }

    @Override
    public void encode(ByteBuf buffer, Object message) throws IOException {
        if (message instanceof Request) {
            encodeRequest(buffer, (Request) message);
        } else if (message instanceof Response) {
            encodeResponse(buffer, (Response) message);
        } else if (message instanceof Event) {
            encodeEvent(buffer, (Event) message);
        } else {
            throw new IllegalArgumentException("未知的消息类型: " + (message == null ? "encode message is null" : message.getClass()));
        }
    }

    @Override
    public Object decode(ByteBuf buffer) throws IOException {
        // 可读字节
        int readable = buffer.readableBytes();

        // 检查可读字节是否小于 header 长度
        if (readable < HEADER_LENGTH) {
            return DecodeResult.NEED_MORE_INPUT;
        }

        // read header
        byte[] header = new byte[HEADER_LENGTH];
        buffer.readBytes(header);

        // check header magic
        if (header[0] != MAGIC[0] && header[1] != MAGIC[1]) {
            throw new CodecException("不合法的标识头");
        }

        // check body size
        int bodySize = Bytes.bytes2int(header, 12);
        if (bodySize > buffer.readableBytes()) {
            return DecodeResult.NEED_MORE_INPUT;
        }

        // decode
        return decode(buffer, header, bodySize);
    }

    private void encodeRequest(ByteBuf buffer, Request request) throws IOException {
        encode(buffer,
                request.isTwoWay() ? FLAG_REQUEST_TWO_WAY : FLAG_REQUEST,
                (byte) 0,
                request.getId(),
                request.getData());
    }

    private void encodeEvent(ByteBuf buffer, Event event) throws IOException {
        encode(buffer,
                FLAG_EVENT,
                event.getType().value(),
                event.getId(),
                event.getData());
    }

    private void encodeResponse(ByteBuf buffer, Response response) throws IOException {
        encode(buffer,
                FLAG_RESPONSE,
                response.getState(),
                response.getId(),
                response.getState() == Response.OK ? response.getResult() : response.getErrorMessage());
    }

    private Object decode(ByteBuf buffer, byte[] header, int bodySize) throws IOException {
        // 标记位
        byte flag = header[2];

        // 解析body
        Object body = null;
        if (bodySize > 0) {
            body = decodeBody(new ByteBufInputStream(buffer, bodySize));
        }

        switch (flag) {
            case FLAG_RESPONSE:     // 解析响应
                return decodeResponse(header, body);

            case FLAG_REQUEST_TWO_WAY:  // 请求
            case FLAG_REQUEST:          // 请求
                return decodeRequest(header, body);

            case FLAG_EVENT:    // 事件
                return decodeEvent(header, body);

            default:
                throw new CodecException("无效的标记: " + flag);
        }
    }

    private Response decodeResponse(byte[] header, Object body) throws IOException {
        Response res = new Response();
        try {
            byte state = header[3];
            res.setState(state);
            res.setId(Bytes.bytes2long(header, 4));
            if (state == Response.OK) {
                res.setResult(body);
            } else {
                res.setErrorMessage((String) body);
            }
        } catch (Throwable t) {
            res.setState(Response.INTERNAL_ERROR);
            res.setErrorMessage(CommonUtil.toString(t));
        }
        return res;
    }

    private Event decodeEvent(byte[] header, Object body) throws IOException {
        Event event = new Event();
        try {
            event.setId(Bytes.bytes2long(header, 4));
            event.setType(header[3]);
            event.setData(body);
            return event;
        } catch (Throwable t) {
            throw new CodecException("解析[Event]时, 发生一个错误", t);
        }
    }

    private Request decodeRequest(byte[] header, Object body) throws IOException {
        long id = Bytes.bytes2long(header, 4);
        byte flag = header[2];
        try {
            Request request = new Request(id);
            request.setTwoWay(flag == FLAG_REQUEST_TWO_WAY);
            request.setData(body);
            return request;
        } catch (Throwable t) {
            throw new CodecException("解析Request时, 发生一个错误, 消息ID[" + id + "]", t);
        }
    }

    /**
     * 写入数据至<code>buffer</code>
     *
     * @param buffer
     * @param flag      数据标记
     * @param state     响应状态码
     * @param id        消息id
     * @param data      数据
     * @throws IOException
     */
    private void encode(ByteBuf buffer, byte flag, byte state, long id, Object data) throws IOException {
        // 初始可写索引位置
        int savedWriteIndex = buffer.writerIndex();

        // 写入索引跳过 header
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);

        // 报文长度
        int len = 0;
        if (data != null) {
            // 序列化数据, 并获取到序列化数据的长度
            ByteBufOutputStream bos = new ByteBufOutputStream(buffer);
            encodeBody(bos, data);
            len = bos.writtenBytes();
        }

        // 移动到初始 write index
        buffer.writerIndex(savedWriteIndex);

        // 写入 header
        writeHeader(buffer,
                flag,
                state,
                id,
                len);

        // 移动write index为写完数据最终的索引位置
        buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    }

    /**
     * 将header数据写入至<code>buffer</code>
     *
     * @param buffer
     * @param flag          数据标记
     * @param state         响应状态码
     * @param id            消息id
     * @param bodyLength    报文长度
     */
    private void writeHeader(ByteBuf buffer,
                             byte flag,
                             byte state,
                             long id,
                             int bodyLength) {

        buffer.writeBytes(MAGIC);       // 2 字节
        buffer.writeByte(flag);         // 1 字节
        buffer.writeByte(state);        // 1 字节
        buffer.writeLong(id);           // 8 字节
        buffer.writeInt(bodyLength);    // 4 字节
    }

    /**
     * 序列化传输数据
     *
     * @param output    序列化的数据, 写入此流中
     * @param data
     * @return  序列化数据的长度, 用于写入bodySize
     */
    protected void encodeBody(OutputStream output, Object data) throws IOException {
        Serialization serialization = getSerialization();
        serialization.serialize(output, data);
    }

    /**
     * 反序列化传输数据
     *
     * @param input     从此流中读取数据
     * @return
     * @throws IOException
     */
    protected Object decodeBody(InputStream input) throws IOException {
        Serialization serialization = getSerialization();
        return serialization.deserialize(input);
    }

}
